package cn.myframe;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Before;
import org.junit.Test;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.Future;

/*******************
 *@Describe
 *@author 封神梦
 *@Time 2022/4/18
 *@Company http://com.yiview.com
 ****************/
public class KafkaTest {
    public static String TOPIC_NAME = "test3";

    @Before
    public void CreateConnetion() {

    }

    /**
     * 创建主题测试
     */
    @Test
    public void CreateTopicTest() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        // props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // props.put("security.protocol", "SASL_PLAINTEXT");
        // props.put("sasl.kerberos.service.name", "kafka");
        AdminClient clent = AdminClient.create(props);
        CreateTopicsResult ret = clent.createTopics(Arrays.asList(new NewTopic("test1", 4, (short) 1)));
        System.out.println("创建主题结果：" + ret.toString());
    }

    /**
     * 消息生产测试
     */
    @Test
    public void producerTest() {
        Properties properties = new Properties();
        properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        KafkaProducer producer = new KafkaProducer(properties);
        ProducerRecord record = new ProducerRecord("test1", "全球世界大战");
        Future send = producer.send(record);
        producer.close();
    }

    /**
     * 消息读取测试
     */
    @Test
    public void ConsumerTest() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put("host.name", "127.0.0.1");

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        //消费者组
        props.setProperty("group.id", "group1");
        KafkaConsumer consumer = new KafkaConsumer(props);
        //订阅topic
        consumer.subscribe(Arrays.asList("test1"));

        while (true) {
            //拉取数据
            ConsumerRecords poll = consumer.poll(5000);
            ((ConsumerRecords) poll).forEach(
                    data -> {
                        System.out.println(((ConsumerRecord) data).value());
                    }
            );
        }
    }

}
